package Test

/**
 * 龚英杰：
 * 车辆超速信息
 * 统计车辆超速20%的车辆，保存到mysql的g_resource.speedCar表中
 *
 * 表字段如下：
 * car :违规车辆牌号
 * speed ：速度
 * areaId ：区域
 * monitorId ：卡口
 * actionId ：拍摄时间
 *
 */

import org.apache.flink.api.common.eventtime.WatermarkStrategy
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.api.common.state.MapStateDescriptor
import org.apache.flink.api.scala._
import org.apache.flink.configuration.Configuration
import org.apache.flink.connector.kafka.source.KafkaSource
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.flink.streaming.api.functions.source.{RichSourceFunction, SourceFunction}
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.util.Collector

import java.sql.{Connection, DriverManager, PreparedStatement}
import java.util.Properties

object rw1 {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
env.setParallelism(1)
    val stream = env.fromSource(KafkaSource.builder()
    .setTopics("test")
    .setBootstrapServers("192.168.23.62:9092")
    .setValueOnlyDeserializer(new SimpleStringSchema())
    .setStartingOffsets(OffsetsInitializer.earliest()).build()
    ,WatermarkStrategy.noWatermarks()
    ,"test")
      val carInfo = stream.map(a=>{
        val strings = a.split("\t")
        strings
      })

//    读取限速信息
    val speedLimit = env.addSource(new RichSourceFunction[(String,String,String,String)] {
      var conn:Connection = _
      var pst:PreparedStatement = _

      override def open(parameters: Configuration): Unit = {
        println(222)
        Class.forName("com.mysql.jdbc.Driver")
        conn = DriverManager.getConnection("jdbc:mysql://192.168.23.62:3306/flinktest", "root", "123456")
        pst = conn.prepareStatement("select * from speedMax")
      }
      override def run(sourceContext: SourceFunction.SourceContext[(String,String,String,String)]): Unit = {
          val set = pst.executeQuery()
          while(set.next()){
            val areaId = set.getString(1)
            val roadId = set.getString(2)
            val monitorId = set.getString(3)
            val speedLimit = set.getString(4)
            sourceContext.collect((areaId,roadId,monitorId,speedLimit))
          }
      }

      override def cancel(): Unit = {
          conn.close()
          pst.close()
      }
    })

    val mpd = new MapStateDescriptor[String,String]("bcMap",classOf[String],classOf[String])
    val bcLimit = speedLimit.broadcast(mpd)

    val outs = carInfo.connect(bcLimit).process(new BroadcastProcessFunction[Array[String],(String,String,String,String),(String,String,String,String,String)] {
      override def processElement(in1: Array[String], readOnlyContext: BroadcastProcessFunction[Array[String], (String,String,String,String), (String,String,String,String,String)]#ReadOnlyContext, collector: Collector[(String,String,String,String,String)]): Unit = {
        val key =in1(0) + "_" + in1(1) + "_" + in1(2)

        val readOnly = readOnlyContext.getBroadcastState(mpd)
        if(readOnly.contains(key)){
          if(in1(6).toDouble > readOnly.get(key).toDouble * 1.2){
            collector.collect((in1(5),in1(6),in1(0),in1(2),in1(4)))
            println((in1(5),in1(6),in1(0),in1(2),in1(4)))
          }
        }
      }

      override def processBroadcastElement(in2: (String,String,String,String), context: BroadcastProcessFunction[Array[String], (String,String,String,String), (String,String,String,String,String)]#Context, collector: Collector[(String,String,String,String,String)]): Unit = {
        val bcState = context.getBroadcastState(mpd)
        val key = in2._1 +"_" + in2._2 + "_" + in2._3
        bcState.put(key,in2._4)

      }
    })
    outs.print()
    outs.addSink(new RichSinkFunction[(String,String,String,String,String)] {
      var Sinkconn:Connection = _
      var Sinkstatement:PreparedStatement = _
      val sql = "insert into speedCar value(?,?,?,?,?)"
      val user = "root"
      val password = "123456"
      val url = "jdbc:mysql://192.168.23.62:3306/g_resource?useSSL=false&useUnicode=true&characterEncoding=utf8"
      val driver = "com.mysql.jdbc.Driver"


      override def open(parameters: Configuration): Unit = {
        Class.forName(driver)
        Sinkconn = DriverManager.getConnection(url,user,password)
        Sinkstatement = Sinkconn.prepareStatement(sql)
      }

      override def invoke(value: (String, String, String, String, String), context: SinkFunction.Context): Unit = {
        Sinkstatement.setString(1,value._1)
        Sinkstatement.setString(2,value._2)
        Sinkstatement.setString(3,value._3)
        Sinkstatement.setString(4,value._4)
        Sinkstatement.setString(5,value._5)
        Sinkstatement.addBatch()
        Sinkstatement.execute()
      }

    })





    env.execute()
  }
}
